package com.beesphere.products.throttling.processor.async;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.beesphere.products.throttling.AlreadyStoppedException;
import com.beesphere.products.throttling.AsyncCallback;
import com.beesphere.products.throttling.UnitOfWork;

public abstract class DelayAsyncProcessorSupport extends DelegateAsyncProcessor {
	
	private static final Logger logger = LoggerFactory.getLogger (DelayAsyncProcessorSupport.class);
    
	private CountDownLatch stoppedLatch = new CountDownLatch (1);
    private boolean fastStop = true;

    public DelayAsyncProcessorSupport (AsyncProcessor processor) {
        super (processor);
    }

    public boolean process (UnitOfWork unitOfWork, long timeout, AsyncCallback asyncCallback) {
        try {
        	delay (unitOfWork);
        } catch (Throwable th) {
        	unitOfWork.setError (th);
        }
        return super.process (unitOfWork, timeout, asyncCallback);
    }

    public boolean isFastStop() {
        return fastStop;
    }

    /**
     * Enables & disables a fast stop; basically to avoid waiting a possibly
     * long time for delays to complete before the context shuts down; instead
     * the current processing method 
     */
    public void setFastStop(boolean fastStop) {
        this.fastStop = fastStop;
    }

    protected abstract void delay (UnitOfWork unitOfWork) throws Exception;

    /**
     * Wait until the given system time before continuing
     * 
     * @param time the system time to wait for
     * @param exchange the exchange being processed
     */
    protected void waitUntil (long time, UnitOfWork unitOfWork) throws Exception {
        while (true) {
            long delay = time - currentSystemTime();
            if (delay < 0) {
                return;
            } else {
                if (isFastStop() && !isRunAllowed()) {
                    throw new AlreadyStoppedException ();
                }
                try {
                    sleep(delay);
                } catch (InterruptedException e) {
                    handleSleepInteruptedException(e);
                }
            }
        }
    }

    protected void sleep (long delay) throws InterruptedException {
        if (delay <= 0) {
            return;
        }
        if (logger.isDebugEnabled()) {
        	logger.debug ("Sleeping for: " + delay + " millis");
        }
        if (isFastStop()) {
            stoppedLatch.await(delay, TimeUnit.MILLISECONDS);
        } else {
            Thread.sleep(delay);
        }
    }

    /**
     * Called when a sleep is interupted; allows derived classes to handle this
     * case differently
     */
    protected void handleSleepInteruptedException (InterruptedException e) {
        logger.debug("Sleep interupted: " + e, e);
    }

    protected long currentSystemTime () {
        return System.currentTimeMillis ();
    }
    
    protected abstract boolean isRunAllowed ();
    
}
